package com.bfxy.kafka.producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import lombok.extern.slf4j.Slf4j;

/**
 * kafka producer 服务， 封装发送消息的方法
 * @author zhengdp
 */
@Slf4j
@Component
public class KafkaProducerService {

	
	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;

	/**
	 * kafka producer 发送消息。 利用kafkaTemplate 进行消息的发送
	 * @param topic 主题
	 * @param msg 消息
	 */
	public void sendMessage(String topic, Object msg) {
		
		ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, msg);

		// ListenableFutureCallback 异步回调方法，接收一个SendResult对象，SendResult对象中封装了ProducerRecord 和RecordMetadata
		// ProducerRecord<K,V> ==> k : topic , V:msg
		// 关于 RecordMetadata 在Apache 提供的kafka依赖中有看到过，其回调方法就会传入这个参数
		future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
			@Override
			public void onSuccess(SendResult<String, Object> result) {
				log.info("发送消息成功: " + result.toString());
			}

			@Override
			public void onFailure(Throwable throwable) {
				log.error("发送消息失败: " + throwable.getMessage());
			}
		});
		
	}
	
	
	
	
}
